package com.gxljc.bear.crawler.weixin;

import com.mongodb.*;
import com.gxljc.commons.mongo.cli.BaseMongoRepository;
import com.gxljc.commons.util.Pair;
import com.gxljc.commons.util.config.CommConsts;
import com.gxljc.bear.crawler.base.BaseMongoTools;
import com.gxljc.bear.crawler.base.DolphinCrawlerConf;
import com.gxljc.bear.crawler.base.DolphinCrawlerConsts;
import com.gxljc.bear.crawler.util.DateUtil;
import com.gxljc.bear.crawler.util.MailService;
import org.apache.commons.collections.CollectionUtils;
import org.apache.log4j.Logger;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.jsoup.Jsoup;
import org.jsoup.nodes.Document;
import org.jsoup.nodes.Element;
import org.jsoup.select.Elements;

import java.io.Serializable;
import java.util.*;

/**
 * 微信PAGE监控数据爬虫。
 * 统计t_newrank_page,t_sogou_page当天和昨天发表的文章情况
 *
 * @author tanghaitao
 * @since 2022-10-13
 */
public class MonitorPage implements Serializable {
    public static Logger LOG = Logger.getLogger(MonitorPage.class);
    private Boolean proxy = false;

    public MonitorPage() {
    }

    public void monitor() throws Exception {
        Map<String, Integer> todaySogou = todayAccount(WeixinConst.MONGODB_SOGOU_PAGE);
        Map<String, Integer> todayNewRankSeeds = todayAccount(WeixinConst.MONGODB_NEWRANK_PAGE);
        Map<String, Integer> yesterdaySogouSeeds = yesterdayAccount(WeixinConst.MONGODB_SOGOU_PAGE);
        Map<String, Integer> yesterdayNewRankSeeds = yesterdayAccount(WeixinConst.MONGODB_NEWRANK_PAGE);
        String mailContent = genMailContent(todaySogou, todayNewRankSeeds, yesterdaySogouSeeds, yesterdayNewRankSeeds);
        String title = genTitle();
        MailService mail = new MailService();
        mail.sendMail(title, mailContent);
    }

    private String genTitle() {
        DolphinCrawlerConf conf = DolphinCrawlerConf.getInstance();
        String platformName = conf.get(DolphinCrawlerConsts.PARAM_PLATFORM_NAME);
        return platformName + "平台weixin爬取统计--" + DateUtil.dateFormat(new Date());
    }

    private String genMailContent(Map<String, Integer> todaySogou, Map<String, Integer> todayNewRankSeeds, Map<String, Integer> yesterdaySogouSeeds, Map<String, Integer> yesterdayNewRankSeeds) {
        String content = "";
        content = "<b>sogou page:</b><br/>";
        content += listContentSogou(todaySogou, yesterdaySogouSeeds);
        content += "<br/><br/><b>newRank page:</b><br/>";
        content += listContentNewRank(todayNewRankSeeds, yesterdayNewRankSeeds);
        return content;
    }

    private String listContentNewRank(Map<String, Integer> today, Map<String, Integer> yesterday) {
        List<Pair<String, Integer>> list = new LinkedList<>();
        for (Map.Entry<String, Integer> entry : yesterday.entrySet()) {
            list.add(new Pair<String, Integer>(entry.getKey(), entry.getValue()));
        }
        Collections.sort(list, new Comparator<Pair<String, Integer>>() {
            public int compare(Pair<String, Integer> arg0, Pair<String, Integer> arg1) {
                return arg0.getSecond().compareTo(arg1.getSecond());
            }
        });
        return genListContent(list, today, yesterday);
    }

    private String listContentSogou(Map<String, Integer> today, Map<String, Integer> yesterday) {
        List<Pair<String, Integer>> list = new LinkedList<>();
        for (Map.Entry<String, Integer> entry : today.entrySet()) {
            list.add(new Pair<String, Integer>(entry.getKey(), entry.getValue()));
        }
        Collections.sort(list, new Comparator<Pair<String, Integer>>() {
            public int compare(Pair<String, Integer> arg0, Pair<String, Integer> arg1) {
                return arg0.getSecond().compareTo(arg1.getSecond());
            }
        });
        return genListContent(list, today, yesterday);
    }

    private String genListContent(List<Pair<String, Integer>> list, Map<String, Integer> today, Map<String, Integer> yesterday) {
        if (CollectionUtils.isEmpty(list)) return "";
        StringBuffer content = new StringBuffer("<table>");
        content.append("<tr>");
        content.append("<td>账号ID</td><td>今天</td><td>昨天</td>");
        content.append("</tr>");
        for (Pair<String, Integer> l : list) {
            content.append("<tr>");
            content.append("<td>").append(l.first).append("</td><td> ").append(today.get(l.first)).append("</td><td> ").append(yesterday.get(l.first)).append("</td>");
            content.append("</tr>");
        }
        content.append("</table>");
        return content.toString();
    }

    private Map<String, Integer> todayAccount(String tableName) throws Exception {
        Date today = new Date();
        Date tomorrow = DateUtil.addCalendar(today, 1);
        Date todayDate = DateUtil.parseFormat(DateUtil.dateFormat(today, "yyyy-MM-dd"));
        Date tomorrowDate = DateUtil.parseFormat(DateUtil.dateFormat(tomorrow, "yyyy-MM-dd"));
        List<String> accountIds = getDateAccountIds(todayDate, tomorrowDate, tableName);
        return accountCompare(accountIds);
    }

    private Map<String, Integer> yesterdayAccount(String tableName) throws Exception {
        Date today = new Date();
        Date yesterday = DateUtil.minusCalendar(today, 1);
        Date todayDate = DateUtil.parseFormat(DateUtil.dateFormat(today, "yyyy-MM-dd"));
        Date yesterdayDate = DateUtil.parseFormat(DateUtil.dateFormat(yesterday, "yyyy-MM-dd"));
        List<String> accountIds = getDateAccountIds(yesterdayDate, todayDate, tableName);
        return accountCompare(accountIds);
    }

    private Map<String, Integer> accountCompare(List<String> accountIds) throws Exception {
//        if (CollectionUtils.isEmpty(accountIds)) return null;
        Map<String, Integer> all = new HashMap<String, Integer>();
        for (String id : accountIds) {
            Integer num = all.get(id);
            if (num == null) all.put(id, 1);
            else all.put(id, num.intValue() + 1);
        }
        List<AccountTable> accountSeeds = WeixinUtil.getAccountSeed();
        for (AccountTable seed : accountSeeds) {
            String accountId = seed.accountId;
            if (all.get(accountId) == null) all.put(accountId, 0);
        }
        return all;
    }

    //获取seed
    public List<String> getDateAccountIds(Date startDate, Date endDate, String tableName) throws Exception {
        BaseMongoRepository repo = BaseMongoTools.getInstance(DolphinCrawlerConsts.MongoDBName.WEIXIN.getName());
        BasicDBObject keys = new BasicDBObject();
        keys.put(WeixinConst.SCHEMA_COLUMN_NAME.ACCOUNT_ID.getValue(), 1);
        DBObject query = new BasicDBObject();
        BasicDBList values = new BasicDBList();
        BasicDBObject dbo1 = new BasicDBObject();
        BasicDBObject dbo2 = new BasicDBObject();
        dbo1.put(WeixinConst.SCHEMA_COLUMN_NAME.PUBLISH_DATE.getValue(), new BasicDBObject("$gte", startDate));
        dbo2.put(WeixinConst.SCHEMA_COLUMN_NAME.PUBLISH_DATE.getValue(), new BasicDBObject("$lt", endDate));
        values.add(dbo1);
        values.add(dbo2);
        query.put("$and", values);
        DBCursor cursor = repo.getCollection(tableName)
                .find(query, keys)
                .addOption(Bytes.QUERYOPTION_NOTIMEOUT);
        List<String> seeds = new LinkedList<>();
        List<DBObject> dataIterator = cursor.toArray();
        for (DBObject data : dataIterator) {
            try {
                Object accountId = data.get(WeixinConst.SCHEMA_COLUMN_NAME.ACCOUNT_ID.getValue());
                seeds.add(accountId.toString());
            } catch (Exception er) {
                er.printStackTrace();
            }
        }
        return seeds;
    }
}
